package com.rpc.demo.common.zookeeper;

import com.rpc.demo.common.constant.ZookeeperConst;
import org.apache.zookeeper.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @Author: ZhaoJie
 * @Date: 2022/4/6
 * @Description: 服务订阅类提供向注册中心订阅服务的功能, 涉及服务发现与负载均衡.
 * @Modify: ===================================================
 * Modifier        Modifytime        Description
 * ===================================================
 */

public class ServiceSubscribe {
    private static Logger logger = LoggerFactory.getLogger(ServiceSubscribe.class);
    private ZooKeeper zk;
    private List<String> providerList;

    /**
     * @author: zhaojie
     * @date: 2022/4/6
     * @description:
     * @method: ServiceSubscribe
     * @param null
     * @class:  ServiceSubscribe
     * @return:
     */
    public ServiceSubscribe() {
        try {
            CountDownLatch connectedSignal = new CountDownLatch(1);
            zk = new ZooKeeper(ZookeeperConst.host, 10000, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getState() == Event.KeeperState.SyncConnected) {
                        connectedSignal.countDown();
                    }
                }
            });
            //因为监听器是异步操作, 要保证监听器操作先完成, 即要确保先连接上ZooKeeper再返回实例.
            connectedSignal.await();

            //创建dubbo注册中心的根节点(持久节点)
            if (zk.exists(ZookeeperConst.rootNode, false) == null) {
                zk.create(ZookeeperConst.rootNode, "".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }
        } catch (Exception e) {
            logger.error("connect zookeeper server error.", e);
        }
    }


    /**
     * @author: zhaojie 
     * @date: 2022/4/6
     * @description: 
     * @method: subscribe  
     * @param null
     * @class:  ServiceSubscribe 
     * @return: 
     */
    public String subscribe(String serviceName) {
        //服务节点路径
        String servicePath = ZookeeperConst.rootNode + "/" + serviceName;
        try {
            //获取服务节点下的所有子节点, 即服务的RPC地址
            providerList = zk.getChildren(servicePath, new Watcher() {
                @Override
                public void process(WatchedEvent event) {
                    if (event.getType() == Event.EventType.NodeChildrenChanged) {
                        try {
                            //循环监听
                            providerList = zk.getChildren(servicePath, true);
                        } catch (KeeperException | InterruptedException e) {
                            logger.error("Consumer在ZooKeeper订阅服务-注册监听器报错", e);
                        }
                    }
                }
            });
        } catch (Exception e) {
            logger.error("从注册中心获取服务报错.", e);
        }
        logger.info(serviceName + "的服务提供者列表: " + providerList);

        //负载均衡
        RandomLoadBalance randomLoadBalance = new RandomLoadBalance();
        return randomLoadBalance.doSelect(providerList);
    }

}
